#include "AsyncLogQueue.h"
#include <algorithm>
#include <atomic>
#include <cassert>
#include <cstddef>
#include <cstdint>
#include <cstdlib>
#include <cstring>
#include <stdexcept>
#include <unistd.h>
#include "Common.h"

namespace xncc {

class AsyncLogQueueElement
{
  public:
    AsyncLogQueueElement(uint64_t size) : MASK(size - 1)
    {
        if (!is_power_of_2(size)) {
            size = roundup_pow_of_two(size);
        }
        if (UNLIKELY(size == 0)) {
            throw std::logic_error("cant not alloc memory");
        }
        in           = 0;
        out          = 0;
        currReadLen  = 0;
        currWriteLen = 0;
        data         = new char[size];
        assert(data != nullptr);
    }
    ~AsyncLogQueueElement()
    {
        delete[] data;
    };

    auto capacity() const -> uint64_t
    {
        return MASK + 1;
    }

    auto getWriteBuffer(uint64_t len) -> char*
    {
        const auto cached_out  = out;
        const auto read_index  = (cached_out & MASK);
        const auto write_index = (in & MASK);
        const auto unused      = (capacity() - (in - cached_out));
        if (len > unused) {
            return nullptr;
        }
        if ((capacity() - write_index) >= len) {
            currWriteLen = len;
            if ((write_index + len) > currentFinal) {
                currentFinal = capacity();
            }
            return data + write_index;
        }
        if (len <= read_index) {
            currWriteLen = len + (capacity() - write_index);
            currentFinal = write_index;
            return data;
        }
        return nullptr;
    }

    void finishWrite()
    {
        __asm__ __volatile__("" ::: "memory");
        const auto result = (in + currWriteLen);
        in                = result;
    }

    auto getReadBuffer(uint64_t len) -> char*
    {
        const auto cached_in   = in;
        const auto read_index  = (out & MASK);
        const auto write_index = (cached_in & MASK);
        if ((cached_in - out) >= len) {
            __asm__ __volatile__("" ::: "memory");  // prevent new write_index old final
            if (read_index == currentFinal) {
                if (write_index >= len) {
                    currReadLen = len + (capacity() - read_index);
                    return data;
                }
                else {
                    fprintf(stderr, "read  what happened ? \n");
                }
            }
            else {
                currReadLen = len;
                return data + read_index;
            }
        }
        else {
            return nullptr;
        }

        return nullptr;
    }

    void finishRead()
    {
        __asm__ __volatile__("" ::: "memory");
        const auto result = (out + currReadLen);
        out               = result;
    }

    bool empty() const
    {
        const auto cached_in  = in;
        const auto cached_out = out;
        return (cached_in == cached_out);
    }

  private:
    auto unused() const -> uint64_t
    {
        return (MASK + 1) - (in - out);
    }

    static auto is_power_of_2(uint64_t n) -> bool
    {
        return (n != 0 && ((n & (n - 1)) == 0));
    }

    static auto roundup_pow_of_two(uint64_t n) -> uint64_t
    {
        n |= n >> 1;
        n |= n >> 2;
        n |= n >> 4;
        n |= n >> 8;
        n |= n >> 16;
        n |= n >> 32;
        n += 1;
        return n;
    }

    static auto rounddown_pow_of_two(uint64_t n) -> uint64_t
    {
        return roundup_pow_of_two(n) >> 1;
    }

    const uint64_t    MASK;
    volatile uint64_t out;
    volatile uint64_t currReadLen;
    volatile uint64_t in;
    volatile uint64_t currWriteLen;
    volatile uint64_t currentFinal;
    char*             data;
};

namespace {
    std::atomic<int> INIT_QUEUE_ID(1);
}
AsyncLogQueue::AsyncLogQueue(uint64_t size)
    : lastReadIndex_(0), queueId_(INIT_QUEUE_ID.fetch_add(1)), canDestory_(false)
{
    bufferList_[0] = std::make_shared<AsyncLogQueueElement>(size);
    currReadPtr_   = bufferList_[0];
    currWritePtr_  = bufferList_[0];
}

void AsyncLogQueue::updateCurrentWritePtr()
{
    // 写指针的改变 以及使用都是在写线程操作，因此不存在多线程竞争问题
    std::lock_guard<std::mutex> lk(mutex_);
    int                         maxIndex = bufferList_.rbegin()->first;
    std::size_t                 oldCap   = bufferList_.rbegin()->second->capacity();
    maxIndex++;
    auto newSize = oldCap * 2;
    newSize      = std::min(newSize, MAX_SINGLE_QUEUE_LENGTH);
    auto ptr     = std::make_shared<AsyncLogQueueElement>(newSize);  // 扩容
    TRACE_DEBUG("queue %d expand size:%luM", queueId_, newSize / 1024 / 1024);
    bufferList_[maxIndex] = ptr;
    currWritePtr_         = ptr;
}

void AsyncLogQueue::updateCurrentReadPtr()
{
    // 读指针的改变 以及使用都是在读线程操作，因此不存在多线程竞争问题
    std::lock_guard<std::mutex> lk(mutex_);
    int                         originIndex = lastReadIndex_;
    lastReadIndex_++;
    TRACE_DEBUG("queue %d free size:%luM", queueId_, currReadPtr_->capacity() / 1024 / 1024);
    currReadPtr_ = bufferList_[lastReadIndex_];
    bufferList_.erase(originIndex);
}

auto AsyncLogQueue::tryGetWriteBuffer(uint64_t len) -> char*
{
    return currWritePtr_->getWriteBuffer(len);
}

auto AsyncLogQueue::getWriteBuffer(uint64_t len) -> char*
{
    char* buf = currWritePtr_->getWriteBuffer(len);
    if (buf == nullptr) {                          // buffer full failed
        updateCurrentWritePtr();                   // expand
        buf = currWritePtr_->getWriteBuffer(len);  // 只扩容一次
    }
    return buf;
}
void AsyncLogQueue::finishWrite()
{
    currWritePtr_->finishWrite();
}
auto AsyncLogQueue::getReadBuffer(uint64_t len) -> char*
{
    char* buf = currReadPtr_->getReadBuffer(len);
    if (buf == nullptr) {
        bool needUpdate = true;
        // 可能没有数据的时候就开始读了，这个时候自动扩容的情况下 没有扩容情况下 不更新读指针
        std::size_t listSize = 0;
        {
            std::lock_guard<std::mutex> lk(mutex_);
            listSize = bufferList_.size();
        }
        if (listSize == 1) {
            needUpdate = false;
        }
        if (needUpdate) {
            buf = currReadPtr_->getReadBuffer(len);
            if (buf != nullptr) {
                return buf;
            }
            updateCurrentReadPtr();
            buf = currReadPtr_->getReadBuffer(len);
        }
    }

    return buf;
}
void AsyncLogQueue::finishRead()
{
    currReadPtr_->finishRead();
}

bool AsyncLogQueue::empty() const
{
    return currReadPtr_->empty();
}

}  // namespace xncc